package com.guchenbo.spark.core

/**
 * @author guchenbo
 * @date 2022/3/4
 */
object RddCache {

  def main(args: Array[String]): Unit = {
    val sc = SparkUtils.sc("map")
    val rdd = sc.makeRDD(List(1, 2, 3, 4), 2)

    //    map，一个一个处理，串行处理
    val rdd2 = rdd.map(n => {
      println("!!!!!!!!!!!!", n)
      n
    })
    val rdd3 = rdd2.map(n => {
      println("@@@@@@@@@@@@", n)
      n
    })
    rdd3.collect().foreach(println)

    //mapPartitions，一个分区一个分区处理，一次性拿到一个分区的数据，放在内存，然后一起处理
    val rdd4 = rdd.mapPartitions(n => {
      println("!!!!!!!!!!!!", n)
      n.map(_ * 2)
    })
    val rdd5 = rdd4.mapPartitions(n => {
      println("@@@@@@@@@@@@", n)
      n.map(_ * 2)
    })
    rdd5.collect().foreach(println)

    val fi = rdd.mapPartitionsWithIndex((idx, iter) => {
      println("!!!!!!!!!!!!", idx)
      //      iter.foreach(println)
      //      iter.map((_, idx))
      iter
    })
    val fi2 = fi.mapPartitionsWithIndex((idx, iter) => {
      println("@@@@@@@@@@@@", idx)
      //      iter.foreach(println)
      //      iter.map((_, idx))
      iter
    })
    fi2.collect().foreach(println)
  }
}
